Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@crates/taskito-core/src/storage/diesel_common/pubsub.rs`:
- Around line 485-518: Update lease_topic_messages to perform the same
subscription registration and mode validation used by read_topic_messages and
ack_topic_cursor before querying or creating delivery rows. Require
subscription_name to exist and have mode exactly "log"; reject unregistered or
non-log subscriptions through the existing error path, while preserving the
current leasing behavior for valid log subscriptions.
- Around line 485-549: The lease_topic_messages transaction must atomically
claim messages before returning them to prevent concurrent callers from
receiving the same row. Update the topic_messages selection to use row-level
locking with SKIP LOCKED where supported, or an equivalent atomic claim
mechanism, ensuring locks/claims cover the availability check through delivery
upsert while preserving ordering and limit behavior.

In `@crates/taskito-core/src/storage/redis_backend/pubsub.rs`:
- Around line 838-844: Update purge_topic_messages to remove each purged
message’s field from every matching pmdeliv topic/subscription hash, enumerating
the topic’s subscribers before or during XTRIM/XDEL cleanup. Ensure fields for
both trimmed and explicitly deleted stream entries are removed while preserving
existing stream purge behavior.
- Around line 849-863: The lease_topic_messages method must enforce the same
subscription-mode restriction as read_topic_messages and ack_topic_cursor. Load
the subscription’s SubEntry and return the established non-log rejection result
unless entry.mode equals SUBSCRIPTION_MODE_LOG, before leasing any messages;
preserve the existing behavior for valid log subscriptions.
- Around line 865-870: Update lease_topic_messages so its XRANGE scan is
COUNT-bounded and paginated rather than issuing an unbounded “-” to “+” query.
Loop through pages with a fixed per-call cap, advancing the range cursor while
skipping in-flight entries, and stop once enough leaseable messages are
collected or the cap/stream end is reached.
- Around line 849-892: The lease_topic_messages method must make availability
checks and lease writes atomic across concurrent callers. Replace the snapshot
HGETALL plus per-entry HSET flow with a Redis transaction or Lua script using
WATCH/MULTI/EXEC, ensuring each message is rechecked against the current
delivery state before writing its lease and retried on conflicts; preserve the
limit, expiry, acknowledged-message, and result-ordering behavior.

In `@crates/taskito-core/tests/rust/storage_tests.rs`:
- Around line 1488-1492: Update the Redis implementation of purge_topic_messages
to compact per-message entries acknowledged by every relevant subscriber, rather
than relying only on log cursors. Remove each fully acknowledged message from
the Redis message storage and delete its associated per-message delivery-state
entries, while preserving unacknowledged messages such as m1 and returning the
number removed.

In `@docs/content/docs/java/api-reference/queue/pubsub.mdx`:
- Around line 23-24: The Java documentation at
docs/content/docs/java/api-reference/queue/pubsub.mdx lines 23-24 and Python
documentation at docs/content/docs/python/api-reference/queue/pubsub.mdx lines
176-201 must consistently describe TopicMessage as returned by both read and
lease APIs. Clarify that cursor-read message IDs use ackTopic/ack_topic, while
leased message IDs use ackMessage/nackMessage or ack_message/nack_message,
including the relevant read_topic, lease_topic, and acknowledgment API
relationships.

In `@docs/content/docs/shared/guides/core/pubsub.mdx`:
- Around line 612-616: Update the lease visibility wording in the pubsub guide
to scope an unexpired lease to the subscription: other lease reads on that same
subscription skip the message until the visibility window expires, while
separate subscriptions may still read it. Preserve the existing ack, nack, and
lease-timeout behavior descriptions.
- Around line 661-665: Update the retention guidance in the subscription cleanup
section to state that Diesel ack-based compaction applies only when a topic is
consumed exclusively per-message; when cursor and per-message subscribers are
mixed, document expiration-based cleanup instead. Preserve the existing Redis
expiration/stream-trim guidance and the warning against mixing consumption
styles.

---

Outside diff comments:
In `@crates/taskito-core/src/storage/diesel_common/pubsub.rs`:
- Around line 371-457: Update unsubscribe() to delete all topic_deliveries rows
for the specified topic and subscription_name before deleting the
topic_subscriptions record, while preserving its existing boolean result based
on subscription deletion. This prevents stale delivery entries from being
included in the per-message compaction denominator.